package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.StockBean;
import cn.itcast.function.KeyFunction;
import cn.itcast.function.StockMinWindowFunction;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.util.DorisUtil;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 个股分时行情
 * 1.每1分钟生成一条最新时间的数据
 */
public class StockMinTask implements ProcessDataInterface {
    @Override
    public void process(DataStream<CleanBean> waterData) {
        /**
         * 业务:每只股票,每分钟获取最新时间(第59s)的一条数据
         * 开发步骤：
         * 1.定义侧边流,分流数据->分表存储
         * 2.数据分组
         * 3.划分时间窗口(60)
         * 4.分时数据处理（新建分时窗口函数）
         * 5.数据分流
         * 6.分表存储(写入Doris)
         */
        //1.定义侧边流,分流数据->分表存储
        //侧边流也叫侧输出流，目的是把主数据流，分成多份
        OutputTag<StockBean> opt = new OutputTag<>("opt", TypeInformation.of(StockBean.class));

        //2.数据分组
        SingleOutputStreamOperator<StockBean> result = waterData.keyBy(new KeyFunction())
                //3.划分时间窗口(60)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //4.分时数据处理（新建分时窗口函数）
                .apply(new StockMinWindowFunction())
                //5.数据分流
                .process(new ProcessFunction<StockBean, StockBean>() {
                    @Override
                    public void processElement(StockBean value, ProcessFunction<StockBean, StockBean>.Context ctx, Collector<StockBean> out) throws Exception {

                        if (value.getSource().equals("sse")) {
                            out.collect(value); //主数据流收集沪市行情数据
                        } else {
                            ctx.output(opt, value); //把深市行情数据封装到侧输出流里面
                        }
                    }
                });

        //6.分表存储(写入Doris)
        //sse
        result.addSink(DorisUtil.getSinkFunction("insert into quot_stock_sse values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));
        //szse
        result.getSideOutput(opt).addSink(DorisUtil.getSinkFunction("insert into quot_stock_szse values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

    }
}
